/********************************************************************
        File :			ServerImpl.cpp
        Creation date :	2010/6/27
		
        License :			Copyright 2010 Ahmed Charfeddine, http://www.pushframework.com

                                   Licensed under the Apache License, Version 2.0 (the "License");
                                   you may not use this file except in compliance with the License.
                                   You may obtain a copy of the License at
				
                                           http://www.apache.org/licenses/LICENSE-2.0
				
                                   Unless required by applicable law or agreed to in writing, software
                                   distributed under the License is distributed on an "AS IS" BASIS,
                                   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
                                   See the License for the specific language governing permissions and
                                   limitations under the License.
	
	
 *********************************************************************/
#include "StdAfx.h"
#include "ServerImpl.h"

#include "Demultiplexor.h"
#include "Dispatcher.h"
#include "ServerStats.h"
#include "IOCPQueue.h"
#include "Acceptor.h"
#include "MonitorAcceptor.h"
#include "../include/BroadcastManager.h"
#include "ChannelFactory.h"
#include "MonitorProtocol.h"
#include "../include/ClientFactory.h"
#include "ClientFactoryImpl.h"
#include "BroadcastManagerImpl.h"
#include "../include/Server.h"




namespace PushFramework {

    void ServerImpl::setProtocol(Protocol* pProtocol) {
        this->pProtocol = pProtocol;
    }

    void ServerImpl::setClientFactory(ClientFactory* pClientFactory) {
        this->pClientFactory = pClientFactory;
        pClientFactory->pImpl->setServerImpl(this);
    }

    void ServerImpl::setWorkerCount(int workerCount) {
        this->workerCount = workerCount;
    }

    ClientFactory* ServerImpl::getClientFactory() {
        return pClientFactory;
    }

    BroadcastManager* ServerImpl::getBroadcastManager() {
        return pBroadcastManager;
    }

    Dispatcher* ServerImpl::getDispatcher() {
        return pDispatcher;
    }

    unsigned int ServerImpl::getSamplingRate() {
        return samplingRate;
    }

    ServerStats* ServerImpl::getStats() {
        return pStats;
    }

    ChannelFactory* ServerImpl::getChannelFactory() {
        return pChannelFactory;
    }

    IOCPQueue* ServerImpl::getIOCPQueue() {
        return pIOCPQueue;
    }

    std::string ServerImpl::getMonitorPassword() {
        return password;
    }

    void ServerImpl::eternal_loop() {


        int sec = 1;
        struct timespec ts;
        pthread_mutex_t cs;
        pthread_mutex_init(&cs, NULL);
        pthread_mutex_lock(&cs);
        pthread_cond_init(&abortCond, NULL);

        int nSecsGCCount = 0;
        int nSecsPerfObservCount = 0;



        while (true) {
            ::clock_gettime(CLOCK_REALTIME, &ts);
            ts.tv_sec += sec;

            int rc = pthread_cond_timedwait(&abortCond, &cs, &ts);
            if (rc == ETIMEDOUT) {

                //Now inc and test if we have to do a job.

                nSecsGCCount++;
                nSecsPerfObservCount++;

                if (nSecsGCCount == uGCPeriod) {
                    nSecsGCCount = 0;
                    pDispatcher->OnStartGC();
                    //Beep( 750, 300 );
                }

                if (nSecsPerfObservCount == samplingRate && isProfilingEnabled) {
                    nSecsPerfObservCount = 0;
                    pDispatcher->OnStartProfiling();
                }
            } else {
                //Server is to die.
                break;
            }

        }


        pthread_mutex_unlock(&cs);

    }

    void* ServerImpl::threadProc(void* WorkContext) {
        ServerImpl* pMe = reinterpret_cast<ServerImpl*> (WorkContext);
        pMe->eternal_loop();
        return 0;
    }

    void ServerImpl::start(bool startInSeparateThread) {


        cout << "MyPath is : " << Utilities::getPath() << endl;

        if (pClientFactory == NULL)
            throw Exception("Fatal : no ClientFactory instance was set");

        assignDefaultProtocol();

        if (!pIOCPQueue->Create())
            throw Exception("Internal problem : unable to create iocp queue");

        if (!pDemux->start()) {
            pIOCPQueue->Free();
            throw Exception("Internal problem : unable to start demultiplexor");
        }

        if (!startListeners()) {
            stopListeners();
            pDemux->stop();
            pIOCPQueue->Free();
        }


        if (isMonitorEnabled && !pMonitorAcceptor->startListening()) {
            stopListeners();
            pDemux->stop();
            pIOCPQueue->Free();
            throw Exception("Internal problem : unable to start monitoring acceptor");
        }

        if (startInSeparateThread) {
            pthread_create(&hThread, NULL, threadProc, (void*) this);
        } else
            eternal_loop();

    }

    void ServerImpl::stop() {
        pthread_cond_signal(&abortCond);
        //pthread_join(hThread, NULL);



        if (isMonitorEnabled)
            pMonitorAcceptor->stopListening();


        stopListeners();


        pDemux->stop();

        pBroadcastManager->pImpl->disposeAllPackets();

        pIOCPQueue->Free();
    }

    void ServerImpl::registerService(unsigned int serviceId, Service* pService, const char* serviceName) {
        pDispatcher->registerService(serviceId, pService, serviceName);
    }

    ServerImpl::ServerImpl(Server* pFacade) {
        this->pFacade = pFacade;

        pProtocol = NULL;
        pClientFactory = NULL;
        listeningPort = 2010;

        workerCount = sysconf(_SC_NPROCESSORS_ONLN) * 2;


        isMonitorEnabled = false;
        isProfilingEnabled = false;
        monitorPort = 2011;
        samplingRate = 10;
        password = "";

        pIOCPQueue = new IOCPQueue();
        pDemux = new Demultiplexor(this);
        pChannelFactory = new ChannelFactory(this);
        pDispatcher = new Dispatcher(this);

        pMonitorAcceptor = new MonitorAcceptor(this);
        pStats = new ServerStats(this);
        pBroadcastManager = new BroadcastManager();
        pBroadcastManager->pImpl->setServerImpl(this);

        uGCPeriod = 10;
    }

    ServerImpl::~ServerImpl(void) {
        delete pIOCPQueue;
        delete pDemux;
        delete pDispatcher;
        delete pChannelFactory;

        delete pMonitorAcceptor;
        delete pStats;
        delete pBroadcastManager;
    }

    Server* ServerImpl::getFacade() {
        return pFacade;
    }

    void ServerImpl::enableRemoteMonitor(short sPort, const char* password) {
        this->monitorPort = sPort;
        this->password = password;
        isMonitorEnabled = true;
        pMonitorAcceptor->setListeningPort(monitorPort);
    }

    void ServerImpl::enableProfiling(unsigned int samplingRate) {
        if (samplingRate >= 0) {
            this->samplingRate = samplingRate;
        }

        isProfilingEnabled = true;
    }

    void ServerImpl::disableProfiling() {
        isProfilingEnabled = false;
    }

    bool ServerImpl::getProfilingStatus() {
        return isProfilingEnabled;
    }

    void ServerImpl::setServeInfos(const char* serverInfos) {
        this->serverInfos = serverInfos;
    }

    std::string ServerImpl::getServerInfos() {
        return serverInfos;
    }

    void ServerImpl::setLoginExpiryDuration(unsigned int uLoginExpireDuration) {
        pChannelFactory->setLoginExpiryDuration(uLoginExpireDuration);
    }

    void ServerImpl::pause() {
        stopListeners();
    }

    void ServerImpl::resume() {
        startListeners();
    }

    void ServerImpl::getVersion(Version& ver) {
        ver.major = 1;
        ver.minor = 0;
        ver.monitorProtocolVer = 1;
    }

    int ServerImpl::getWorkerCount() {
        return workerCount;
    }

    BroadcastManagerImpl* ServerImpl::getBroadcastManagerImpl() {
        return pBroadcastManager->pImpl;
    }

    bool ServerImpl::createListener(int port, ListenerOptions* pOptions /*= NULL*/) {
        listenersMapT::iterator it = listenersMap.find(port);
        if (it != listenersMap.end())
            return false;

        Acceptor* pAcceptor = new Acceptor(this);
        pAcceptor->setListeningPort(port);

        if (pOptions)
            pAcceptor->getOptions() = *pOptions;

        listenersMap[port] = pAcceptor;
        return true;
    }

    bool ServerImpl::startListeners() {
        for (listenersMapT::iterator it = listenersMap.begin();
                it != listenersMap.end();
                it++) {
            Acceptor* pAcceptor = it->second;
            //
            //Start it :
            if (!pAcceptor->startListening()) {
                return false;

            }
        }
        return true;
    }

    void ServerImpl::assignDefaultProtocol() {
        for (listenersMapT::iterator it = listenersMap.begin();
                it != listenersMap.end();
                it++) {
            Acceptor* pAcceptor = it->second;
            //
            if (pAcceptor->getOptions().pProtocol == NULL) {
                if (pProtocol == NULL) {
                    std::string strError = "FATAL : no protocol instance for set for an acceptor";
                    throw Exception(strError);
                }
                pAcceptor->getOptions().pProtocol = pProtocol;
            }
        }
    }

    void ServerImpl::stopListeners() {
        for (listenersMapT::iterator it = listenersMap.begin();
                it != listenersMap.end();
                it++) {
            Acceptor* pAcceptor = it->second;
            pAcceptor->startListening();
        }
    }

    void ServerImpl::setMaxConnections(int nMaxConnections) {
        pChannelFactory->setMaxAllowedConnections(nMaxConnections);
    }

    void ServerImpl::EncodeOnAllProtocols(OutgoingPacket& outgoingPacket) {
        for (listenersMapT::iterator it = listenersMap.begin();
                it != listenersMap.end();
                it++) {
            Protocol* pProtocol = it->second->getOptions().pProtocol;
            pProtocol->encodeOutgoingPacket(outgoingPacket);
        }
    }

}

